#define _GNU_SOURCE
#include "asyncpages.h"
#include "dfruntime_private.h"
#include "page_management.h"
#include "memory_management.h"
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <fcntl.h>
#include <errno.h>

#include "memory_management.h"


#define BUFFER_SIZE 65536 // Max UDP Packet size.
static int multicast_fd = -1;
static bool __inited = false;
static struct sockaddr_in saddr_output;
static struct sockaddr_in saddr_input;
static socklen_t socklen = sizeof(struct sockaddr_in);

// TLS reception buffer.
static __thread char buffer[BUFFER_SIZE];

typedef long long SerialType;
static SerialType *serials;
static SerialType local_serial;




static bool get_next_message( struct dfrt_mem_release_mesg * msg ) {
    return dfrt_mem_next_rls(msg);

}

void async_process_all_messages() {
    static volatile int working = 0;
    if ( !__sync_bool_compare_and_swap( &working, 0, 1 )) {
        // Another task processes messages
        DEBUG("Skipping release message processing.");
        return;
    }

    struct dfrt_mem_release_mesg msg;
    while ( get_next_message(&msg) ) {
        bool commit_changes = false;
#pragma omp critical (page_man)
        {
            // Notify page management
            if ( page_is_wanted(msg.pageaddr) ) {
                struct pagedir_elt pde;
                page_get_direlt(msg.pageaddr, &pde);

                // TODO : set exclusive bit correctly
                if ( dfrt_mem_async_acquire(msg.pageaddr,
                                            msg.size,
                                            true)) {

                    DEBUG("Pagerelease taken over (%p,%lu)",
                          msg.pageaddr,
                          (unsigned long) msg.size );
                    page_managed_acquire(msg.pageaddr,msg.size,true);
                    commit_changes = true;
                } else {
                    // This happens for instance when a page is
                    // released twice before messages are processed
                }
            }
        }

        // Trigger hooks
        if( commit_changes) {
            hook_at_page_arrival(msg.pageaddr);
        }


    }
    working = 0;

}

void async_page_management_init() {


    __inited = true;

}



bool async_request_page( void * page, size_t size, bool exclusive ) {
    // TODO : rewrite this : messy.

    bool acquire_already_ok = false;
    bool acquire_success = false;

#pragma omp critical (page_man)
    {
        // Check if page is already acquired.
        struct pagedir_elt *pagedesc;
        if ( page_is_acquired(page) ) {
            // Then increase the use count :
            page_incr_count(page,size,exclusive,true);
            acquire_already_ok = true;
        } else {
            // Set this page as wanted.
            page_incr_count(page,size,exclusive,false);
        }

        if ( acquire_already_ok ) {

        } else {
            // Try to have it.
            acquire_success = dfrt_mem_async_acquire(page,
                                                     size,
                                                     exclusive);

            if ( acquire_success ) {
                // Notify page management
                page_managed_acquire(page,size,exclusive);

            } else {
                // Check that page is wanted :
                CFATAL(!page_is_wanted(page), "Async acquire without want.");
            }

        }
    }


    // Hooks is just to commit changes : it has guarantee that
    // no one will decrease ownership counter to 0 since he has
    // to do it :
    // Trigger hooks : needs to be outside critical section,
    // in the case of a schedule being performed in place ...
    if ( acquire_success ) {
        hook_at_page_arrival(page);
    }

    bool retbool = acquire_success || acquire_already_ok;

    DEBUG( "Acquire retbool (%p): %d", page, retbool);
    return retbool;


}

void safe_release_page( void * page ) {
#pragma omp critical (page_man)

    {
        CFATAL(!page_is_acquired(page), "Releasing non acquired page : %p", page);

        struct pagedir_elt direlt;
        page_get_direlt(page,&direlt);
        _DEBUG( "Safe release page : %p, usecount = %d (will decrease)", page,  direlt.usecount );

        if ( direlt.usecount == 1 ) {
            dfrt_mem_release(page);
        }

        page_dec_count(page, direlt.size,direlt.exclusive);
    }
}

void sync_safe_acquire( void * page, size_t size, bool exclusive ) {


#pragma omp critical (page_man)
    {
        if ( page_is_acquired(page)) {
            page_incr_count(page,size,exclusive, true);
        } else {
            if ( exclusive ) {
               dfrt_mem_blocking_exclusive_acquire(page,size);
            } else {
               dfrt_mem_blocking_shared_acquire(page,size);
            }

            page_incr_count(page, size, exclusive, true);


        }
        struct pagedir_elt direlt;
        page_get_direlt(page,&direlt);
        DEBUG( "Sync safe acquired %p  to : %d ",page, direlt.usecount);
    }
}


